package cn.com.sparkle.paxos.state;

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.config.Configuration;
import cn.com.sparkle.paxos.model.ElectionId;

public class SelfState {
	private final static Logger logger = Logger.getLogger(SelfState.class);
	private volatile ElectionId electionVoteIdBySelf;
	private volatile ElectionId electionPrepareId;
	private ReentrantLock electionPrepareIdLock = new ReentrantLock();
	private Configuration conf;
	private AtomicLong transportToMasterCount = new AtomicLong(0);//for statistic
	private long paxosPipelineProcessCount = 0;//for statistic
	private AtomicLong responseCustomAddRequestCount = new AtomicLong(0);//for statistic
	private AtomicLong responseAdminAddRequestCount = new AtomicLong(0);//for statistic
	private long executeFromStartCount = 0; // for statistic
	private long recoverRecordFromMasterLossCount = 0; //for statistic

	public ElectionId getElectionVoteIdBySelf() {
		return electionVoteIdBySelf;
	}

	public SelfState(Configuration conf) {
		this.conf = conf;
		electionVoteIdBySelf = new ElectionId("", 0, conf.getConfigNodeSet().getVersion());
		electionPrepareId = new ElectionId("", -1, conf.getConfigNodeSet().getVersion());
	}

	public ElectionId casElectionPrepareId(ElectionId id) {
		try {
			electionPrepareIdLock.lock();
			int compareResult = id.compareTo(electionPrepareId);
			if (conf.isDebugLog() && compareResult != 0) {
				logger.debug(id.getIncreaseId() + id.getAddress() + (compareResult == 1 ? ">" : "<") + electionPrepareId.getIncreaseId()
						+ electionPrepareId.getAddress());
			}
			if (compareResult >= 0 && electionPrepareId.getVersion() == id.getVersion()) {
				electionPrepareId = new ElectionId(id.getAddress(), id.getIncreaseId(), id.getVersion());
				return id;
			}
			return electionPrepareId;
		} finally {
			electionPrepareIdLock.unlock();
		}
	}

	public void addTransportToMasterCount() {
		transportToMasterCount.incrementAndGet();
	}

	public void addPaxosPipelineProcessCount() {
		++paxosPipelineProcessCount;
		if (conf.isDebugLog()) {
			if ((paxosPipelineProcessCount) % 1000 == 0) {
				logger.debug("add count " + paxosPipelineProcessCount);
			}
		}
	}

	public void addRecoverRecordFromMasterLossCount() {
		++recoverRecordFromMasterLossCount;
	}

	public void addResponseCustomAddRequestCount() {
		responseCustomAddRequestCount.incrementAndGet();
	}

	public void addResponseAdminAddRequestCount() {
		responseAdminAddRequestCount.incrementAndGet();
	}

	public void addExecuteFromStartCount(int increase) {
		executeFromStartCount += increase;
	}

	public long getRecoverRecordFromMasterLossCount() {
		return recoverRecordFromMasterLossCount;
	}

	public long getTransportToMasterCount() {
		return transportToMasterCount.get();
	}

	public AtomicLong getResponseCustomAddRequestCount() {
		return responseCustomAddRequestCount;
	}

	public AtomicLong getResponseAdminAddRequestCount() {
		return responseAdminAddRequestCount;
	}

	public synchronized long getPaxosPipelineProcessCount() {
		//synchronized for visible for other thread.
		//The reason of not using violate is optimize for add operation in the single thread
		return paxosPipelineProcessCount;
	}

	public synchronized long getExecuteFromStartCount() {
		//synchronized for visible for other thread.
		//The reason of not using violate is optimize for add operation in the single thread
		return executeFromStartCount;
	}

}
